🔴 Exception handler bypasses new update_task method, leaving stream subscribers hanging forever

When run_task or cancel_task raises an exception in _handle_task_operation, the except block at line 58 calls self.storage.update_task(...) directly instead of the new self.update_task(...) method introduced in this PR. The new self.update_task() method (fasta2a/worker.py:60-118) is documented as "the primary method workers should use to update task state" because it both persists the update to storage AND publishes streaming events to the broker via send_stream_event. By calling self.storage.update_task() directly, no TaskStatusUpdateEvent(final=True) is ever published to subscribers. Any active subscribe_to_stream iterator for that task will block indefinitely waiting for a final event, causing the SSE connection to hang.

Was this helpful? React with 👍 or 👎 to provide feedback.

🔴 Fire-and-forget asyncio.create_task loses exceptions and task reference

At fasta2a/task_manager.py:194, asyncio.create_task(self.broker.run_task(broker_params)) creates a background task but discards the returned Task object. If run_task raises an exception, Python emits a "Task exception was never retrieved" warning and the error is silently lost — the streaming client will never know the task failed to even start. The task object is also subject to garbage collection since no reference is held, which could cause the task to be cancelled unexpectedly (per Python docs: "the event loop only keeps weak references to tasks").

In fasta2a/task_manager.py at line 194 in the stream_message method, the asyncio.create_task result is discarded. Store the task reference and add error handling. One approach: store the task in a local variable, and after the subscribe_to_stream loop completes, check if the background task raised an exception (e.g. via task.result() in a try/except). Alternatively, restructure to use a task group (e.g. anyio.create_task_group) to properly manage the background task's lifecycle and propagate exceptions.

Was this helpful? React with 👍 or 👎 to provide feedback.

🚩 activated_extensions parsed but never consumed by downstream code

At fasta2a/applications.py:144-150, the A2A-Extensions header is parsed and stored on request.state.activated_extensions with a comment saying "so workers / handlers can inspect them." However, this state is never passed to TaskManager, Broker, or Worker — those components receive only TaskSendParams which doesn't include activated extensions information. The request.state is scoped to the HTTP request handler and is not accessible from workers. This appears to be scaffolding for future use rather than a bug, but it means extensions have no functional effect currently.

Was this helpful? React with 👍 or 👎 to provide feedback.

The background asyncio.create_task(self.broker.run_task(...)) isn't tracked or awaited. If it raises, you'll get "Task exception was never retrieved" and the SSE generator may keep streaming indefinitely. Consider capturing the task and handling exceptions (e.g., add a done callback that logs and/or publishes a failed final status) and cancel it if the client disconnects.

FastA2A advertises streaming capability via AgentCapabilities.streaming, but _agent_run_endpoint will still accept and serve message/stream even when self.streaming is False. Add a guard that returns an appropriate JSON-RPC error (e.g., MethodNotFound/UnsupportedOperation) when streaming is disabled so behavior matches the advertised capabilities.

The PR metadata focuses on "extensions", but this change also flips the default advertised streaming capability to True (and adds a new message/stream SSE endpoint). If this broader behavior change is intentional, it should be reflected in the PR title/description (or split into a separate PR) to avoid surprising consumers.

In the exception handler, the worker updates storage directly, which bypasses the new Worker.update_task() logic and therefore won't publish a final status-update event to stream subscribers on failures. Use Worker.update_task(..., state='failed') here (and consider capturing/logging the exception) so streaming clients reliably see task termination.

send_stream_event() holds _subscriber_lock while awaiting stream.send(...). If a subscriber is slow (buffer full) this blocks all subscription changes and event sends across all tasks. Consider copying the current subscriber list under the lock, releasing it before awaiting sends, then reacquiring to prune/commit active subscribers.

stream_message() can miss early stream events: it yields the initial task, then schedules broker.run_task(), and only afterwards starts iterating subscribe_to_stream(). If the worker publishes status updates quickly, they can be sent before the subscription is registered and will be dropped. Restructure so the subscription is registered before starting task execution (and ideally before yielding the first event), or change the broker API so subscription registration happens eagerly when subscribe_to_stream() is called.